package cn.migu.music.hangupsms.dbcache;

import java.time.DayOfWeek;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.RateLimiter;

import cn.migu.music.common.redis.IRedisClient;
import cn.migu.music.common.utils.LogUtils;
import cn.migu.music.hangupsms.common.config.HangUpSmsConfig;
import cn.migu.music.hangupsms.common.entity.MaxNumInfo;
import cn.migu.music.hangupsms.common.entity.UserSendMsgInfo;
import cn.migu.music.hangupsms.common.util.LocalDateTimeUtils;
import cn.migu.music.hangupsms.common.util.MyHolder;
import cn.migu.music.hangupsms.repository.HangupMsgSendUserMapper;
import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class SentSmsCacheManager {
    private static final String            REDIS_KEY_DIFF     = "hangup_sms_record_diff";
    private static final String            REDIS_KEY_SUM      = "hangup_sms_record_sum";
    private static final String            REDIS_KEY_HIT      = "hangup_sms_record_hit";
    private static final String            REDIS_KEY_SAVE_SUM = "hangup_sms_record_save_sum";
    private static final String            DB_NOT_SEND        = "hangup_sms_db_not_send";
    private static final String            CACHE_NOT_SEND     = "hangup_sms_cache_not_send";
    private static final String            DB_SEND            = "hangup_sms_db_send";
    private static final String            CACHE_SEND         = "hangup_sms_cache_send";
    private static final String            SECOND_FORMAT      = "yyyyMMddHHmmss";
    private static final String            LOAD_DB_INTERRUPT  = "hangupsms.loaddata.interrupt";
    private static final String            LOAD_DB_LOCK       = "hangupsms.loaddata.lock";
    private static final DateTimeFormatter DFT                = DateTimeFormatter.ofPattern(SECOND_FORMAT);
    @Autowired
    private IRedisClient                   redisClient;
    @Autowired
    private HangUpSmsConfig                hangUpSmsConfig;
    @Autowired
    private HangupMsgSendUserMapper        hangupMsgSendUserMapper;

    /**
     * 在缓存预热期，由于缓存缺少数据，cacheLimit默认为false，所有手机都发送短信， CACHE_SEND/DB_SEND > 1，CACHE_NO_SEND/DB_NOT_SEND < 1
     * 当缓存数据逐步与数据库一致后，CACHE_SEND/DB_SEND和CACHE_NO_SEND/DB_NOT_SEND趋于等于1
     * 
     * @param dbLimit
     * @param cacheLimit
     * @param cacheRet
     * @param phone
     * @param timeList
     */
    public void staticCacheAndDbCompare(boolean dbLimit, boolean cacheLimit, int cacheRet, String phone, List<LocalDateTime> timeList) {
        if (!shouldSaveRecord(phone)) {
            return;
        }
        // 当缓存为空，db不为空的时候，统计，统计数据不在发生变化，说明数据一致了
        if (cacheRet == -1 && CollectionUtils.isNotEmpty(timeList)) {
            redisClient.incr(REDIS_KEY_DIFF);
        }
        if (dbLimit != cacheLimit) {
            LogUtils.info(log, "different decision:" + phone + ", dbLimit:" + dbLimit + ", cacheLimit:" + cacheLimit);
        }
        if (dbLimit) {
            redisClient.incr(DB_NOT_SEND);
        } else {
            redisClient.incr(DB_SEND);
        }
        if (cacheLimit) {
            redisClient.incr(CACHE_NOT_SEND);
        } else {
            redisClient.incr(CACHE_SEND);
        }
    }

    public void saveSendRecordList(String callerPhone, List<LocalDateTime> timeList) {
        if (!shouldSaveRecord(callerPhone)) {
            return;
        }
        String key = hangUpSmsConfig.getSendRecordCachePrefix() + callerPhone;

        LocalDateTime nearestTime = null;
        int index = 0;
        LocalDateTime now = LocalDateTime.now();
        List<LocalDateTime> timeListValid = new ArrayList<>();
        int monthNow = now.getMonthValue();
        for (LocalDateTime localDateTime : timeList) {
            int month = localDateTime.getMonthValue();
            if (month != monthNow) {
                continue;
            }
            timeListValid.add(localDateTime);
            index++;
            if (nearestTime == null) {
                nearestTime = localDateTime;
                continue;
            }
            if (localDateTime.isAfter(nearestTime)) {
                nearestTime = localDateTime;
            }
        }
        if (CollectionUtils.isEmpty(timeListValid)) {
            return;
        }

        String[] timeArray = new String[index];
        timeListValid.stream().map(localDateTime -> {
            return localDateTime.format(DFT);
        }).collect(Collectors.toList()).toArray(timeArray);
        redisClient.sadd(key, timeArray);
        setExpire(key, nearestTime == null ? LocalDateTime.now() : nearestTime);
        redisClient.incrBy(REDIS_KEY_SAVE_SUM, index);
        if (ThreadLocalRandom.current().nextInt(100) < hangUpSmsConfig.getSendRecordCacheLogRatio()) {
            LogUtils.info(log, "cached phone:" + callerPhone);
        }
    }

    public void saveSendRecord(String callerPhone, LocalDateTime now) {
        if (shouldSaveRecord(callerPhone)) {
            String key = hangUpSmsConfig.getSendRecordCachePrefix() + callerPhone;
            redisClient.sadd(key, now.format(DFT));
            setExpire(key, now);
            redisClient.incr(REDIS_KEY_SAVE_SUM);
            if (ThreadLocalRandom.current().nextInt(100) < hangUpSmsConfig.getSendRecordCacheLogRatio()) {
                LogUtils.info(log, "cached phone:" + callerPhone);
            }
        }
    }

    private boolean shouldSaveRecord(String phone) {
        if (CollectionUtils.isEmpty(hangUpSmsConfig.getSendRecordCacheModeList())) {
            return false;
        }
        long phoneNumber = 0;
        try {
            phoneNumber = Long.parseLong(phone);
        } catch (NumberFormatException e) {
            LogUtils.error(e, log, "illegal phone:%s", phone);
            return false;
        }
        long mode = phoneNumber % hangUpSmsConfig.getSendRecordCacheModeFactor();
        return hangUpSmsConfig.getSendRecordCacheModeList().contains((int) mode);
    }

    private boolean isWeekCrossMonth(LocalDateTime now, MyHolder<Integer> weekendMonthDay, MyHolder<Integer> weekRemainDay) {
        int nowMonth = now.getMonthValue();
        DayOfWeek dayOfWeek = now.getDayOfWeek();
        int amountToAdd = 7 - dayOfWeek.getValue();
        if (weekRemainDay != null) {
            weekRemainDay.set(amountToAdd);
        }
        LocalDateTime future = now.plusDays(amountToAdd);
        int futureMonth = future.getMonthValue();
        if (weekendMonthDay != null) {
            weekendMonthDay.set(future.getDayOfMonth());
        }
        // 模12范围0-11，月份都需要减1
        // 当月下一月：（nowMonth - 1 + 1 + 12）% 12 化简
        return (nowMonth + 12) % 12 == (futureMonth - 1);
    }

    public void setExpire(String key, LocalDateTime now) {
        redisClient.expire(key, getMonthExpire());
        int day = now.getDayOfMonth();
        // 上一个月数据应该删除，我们在月前10天持续关注每个被延期的键是否包含上月应该删除的数据
        if (day > 10) {
            return;
        }
        int nowMonth = now.getMonthValue();
        Set<String> timeList = redisClient.smembers(key);
        List<String> timeListToDelete = new ArrayList<>();
        String nowMonthStr = String.valueOf(nowMonth);
        if (nowMonthStr.length() == 1) {
            nowMonthStr = '0' + nowMonthStr;
        }
        for (String timeStr : timeList) {
            String monthStr = timeStr.substring(4, 6);
            if (StringUtils.equals(nowMonthStr, monthStr)) {
                continue;
            }
            LocalDateTime sendTime = LocalDateTime.parse(timeStr, DFT);
            MyHolder<Integer> weekendMonthDay = new MyHolder<>();
            boolean crossMonth = isWeekCrossMonth(sendTime, weekendMonthDay, null);
            if (!crossMonth) {
                timeListToDelete.add(timeStr);
                continue;
            }
            if (weekendMonthDay.get() < day) {
                timeListToDelete.add(timeStr);
            }
        }
        if (CollectionUtils.isNotEmpty(timeListToDelete)) {
            String[] timeArray = new String[timeListToDelete.size()];
            timeListToDelete.toArray(timeArray);
            redisClient.srem(key, timeArray);
        }
    }

    private int getMonthExpire() {
        Calendar cal = Calendar.getInstance();
        long nowSecond = cal.getTimeInMillis() / 1000;
        // 获取某月最大天数
        int lastDay = cal.getActualMaximum(Calendar.DAY_OF_MONTH);
        // 设置日历中月份的最大天数
        cal.set(Calendar.DAY_OF_MONTH, lastDay);
        cal.set(Calendar.HOUR_OF_DAY, 23);
        cal.set(Calendar.MINUTE, 59);
        cal.set(Calendar.SECOND, 59);
        long futureSecond = cal.getTimeInMillis() / 1000;
        // 让数据暂存7天，防止跨月时候大量请求访问数据库
        // 让数据在3天随机失效，避免数据集中失效
        return (int) (futureSecond - nowSecond) + 7 * 3600 * 24 + ThreadLocalRandom.current().nextInt(3600 * 24 * 3);
    }

    public int isSentLimit(String callerPhone, LocalDateTime now, MaxNumInfo maxNumInfo) {
        redisClient.incrBy(REDIS_KEY_SUM, 1);
        String key = hangUpSmsConfig.getSendRecordCachePrefix() + callerPhone;

        Set<String> timeList = redisClient.smembers(key);
        if (CollectionUtils.isEmpty(timeList)) {
            return -1;
        }
        redisClient.incrBy(REDIS_KEY_HIT, 1);
        LocalDateTime startDayOfMonth = LocalDateTimeUtils.getStartOrEndDayOfMonth(now.toLocalDate(), true);
        LocalDateTime startDayOfWeek = LocalDateTimeUtils.getStartOrEndDayOfWeek(now.toLocalDate(), true);
        LocalDateTime startDayOfDay = LocalDateTimeUtils.getStartOrEndDayOfDay(now.toLocalDate(), true);
        int ret;
        if (exceedMaxCount(timeList, startDayOfMonth, maxNumInfo.getMaxMonthNum())) {
            ret = 1;
        } else if (exceedMaxCount(timeList, startDayOfWeek, maxNumInfo.getMaxWeekNum())) {
            ret = 1;
        } else if (exceedMaxCount(timeList, startDayOfDay, maxNumInfo.getMaxDayNum())) {
            ret = 1;
        } else {
            ret = 0;
        }
        if (ThreadLocalRandom.current().nextInt(100) < hangUpSmsConfig.getSendRecordCacheLogRatio()) {
            LogUtils.info(log, "hit cache:" + JSON.toJSONString(timeList) + ", ret:" + ret);
        }
        return ret;
    }

    private boolean exceedMaxCount(Set<String> sendTimeList, LocalDateTime rangeStart, int maxCount) {
        int sum = 0;
        for (String timeStr : sendTimeList) {
            LocalDateTime sendTime = LocalDateTime.parse(timeStr, DFT);
            if (sendTime.isAfter(rangeStart)) {
                sum++;
                if (sum >= maxCount) {
                    return true;
                }
            }
        }
        return false;
    }

    public Map<String, String> getStatic() {
        Map<String, String> staticMap = new HashMap<>();
        float sumNumber = 0;
        float hitNumber = 0;
        float hitRate = 0;
        float dbSendNumber = 0;
        float dbNotSendNumber = 0;
        float cacheSendNumber = 0;
        float cacheNotSendNumber = 0;
        float sendCorrectRatio = 0;
        float notSendCorrectRatio = 0;
        long savedSumNumber = 0;
        try {
            String sum = redisClient.get(REDIS_KEY_SUM);
            String hit = redisClient.get(REDIS_KEY_HIT);

            sumNumber = Long.parseLong(sum);
            hitNumber = Long.parseLong(hit);
            hitRate = hitNumber / sumNumber;

        } catch (NumberFormatException e) {
            LogUtils.info(e, log, "no hit static");
        }

        try {
            String dbSend = redisClient.get(DB_SEND);
            String cacheSend = redisClient.get(CACHE_SEND);

            dbSendNumber = Long.parseLong(dbSend);
            cacheSendNumber = Long.parseLong(cacheSend);
        } catch (NumberFormatException e) {
            LogUtils.info(e, log, "no send static");
        }

        try {
            String dbNotSend = redisClient.get(DB_NOT_SEND);
            String cacheNotSend = redisClient.get(CACHE_NOT_SEND);
            dbNotSendNumber = Long.parseLong(dbNotSend);
            cacheNotSendNumber = Long.parseLong(cacheNotSend);
        } catch (NumberFormatException e) {
            LogUtils.info(e, log, "no no-send static");
        }

        try {
            String savedSum = redisClient.get(REDIS_KEY_SAVE_SUM);
            if (savedSum != null) {
                savedSumNumber = Long.parseLong(savedSum);
            }
        } catch (NumberFormatException e) {
            LogUtils.info(e, log, "no saved cache sum static");
        }
        long diffNumber = 0;
        try {
            String diff = redisClient.get(REDIS_KEY_DIFF);
            if (diff != null) {
                diffNumber = Long.parseLong(diff);
            }
        } catch (NumberFormatException e) {
            LogUtils.info(e, log, "no db and cache diff static");
        }

        if (dbSendNumber > 0) {
            sendCorrectRatio = cacheSendNumber / dbSendNumber;
        } else {
            sendCorrectRatio = cacheSendNumber == dbSendNumber ? 1 : 0;
        }
        if (dbNotSendNumber > 0) {
            notSendCorrectRatio = cacheNotSendNumber / dbNotSendNumber;
        } else {
            notSendCorrectRatio = cacheNotSendNumber == dbNotSendNumber ? 1 : 0;
        }

        staticMap.put("sum", String.valueOf(sumNumber));
        staticMap.put("hit", String.valueOf(hitNumber));
        staticMap.put("hitRate", String.valueOf(hitRate));

        staticMap.put("dbSend", String.valueOf(dbSendNumber));
        staticMap.put("dbNotSend", String.valueOf(dbNotSendNumber));
        staticMap.put("cacheSend", String.valueOf(cacheSendNumber));
        staticMap.put("cacheNotSend", String.valueOf(cacheNotSendNumber));
        staticMap.put("sendCorrectRatio", String.valueOf(sendCorrectRatio));
        staticMap.put("notSendCorrectRatio", String.valueOf(notSendCorrectRatio));

        staticMap.put("sendRecordCacheSwitch", String.valueOf(hangUpSmsConfig.isSendRecordCacheSwitch()));
        staticMap.put("sendRecordCacheWarmup", String.valueOf(hangUpSmsConfig.isSendRecordCacheWarmup()));
        staticMap.put("savedSum", String.valueOf(savedSumNumber));
        staticMap.put("diffCount", String.valueOf(diffNumber));
        return staticMap;
    }

    public void resetStatic() {
        redisClient.del(REDIS_KEY_HIT, REDIS_KEY_SUM);
        redisClient.del(DB_NOT_SEND, CACHE_NOT_SEND);
        redisClient.del(DB_SEND, CACHE_SEND);
        redisClient.del(REDIS_KEY_DIFF);
        redisClient.del(REDIS_KEY_SAVE_SUM);
    }

    public Set<String> getRedisValue(String phone) {
        String key = hangUpSmsConfig.getSendRecordCachePrefix() + phone;
        Set<String> timeList = redisClient.smembers(key);
        return timeList;
    }

    public void interruptLoadData() {
        redisClient.set(LOAD_DB_INTERRUPT, "1");
    }

    public void deleteLock() {
        redisClient.del(LOAD_DB_LOCK);
    }

    public String getLock() {
        return redisClient.get(LOAD_DB_LOCK);
    }

    @Async("hangupSmsAsyncExecutor")
    public void loadDbData() {
        boolean lock = redisClient.setnxExpire(LOAD_DB_LOCK, "1", 3600 * 3);
        if (!lock) {
            LogUtils.info(log, "[LoadCacheFromDb], load data from db is running ...");
            return;
        }
        redisClient.del(LOAD_DB_INTERRUPT);
        Stopwatch stopwatch = Stopwatch.createStarted();
        for (Map.Entry<String, String> entry : hangUpSmsConfig.getProvinceAndLocalIdMap().entrySet()) {
            String interrupt = redisClient.get(LOAD_DB_INTERRUPT);
            if (hangUpSmsConfig.isSendRecordCacheDbInterrupt() || StringUtils.equals("1", interrupt)) {
                LogUtils.info(log,
                    "[LoadCacheFromDb], load data from db is interrupted. costTime=" + stopwatch.elapsed(TimeUnit.SECONDS) + " second.");
                break;
            }
            int lid = Integer.parseInt(entry.getValue());
            LogUtils.info(log, "[LoadCacheFromDb], load data from db:" + entry.getKey() + "," + entry.getValue());
            loadOneProvinceData(lid);
        }
        LogUtils.info(log, "load data from db is finished. costTime=" + stopwatch.elapsed(TimeUnit.SECONDS) + " second.");
        redisClient.del(LOAD_DB_LOCK);
    }

    private void loadOneProvinceData(int lid) {
        int pageSize = hangUpSmsConfig.getSendRecordCacheDbSize();
        int pageIndex = 0;
        int returnSize = 0;
        RateLimiter rateLimiter = RateLimiter.create(hangUpSmsConfig.getSendRecordCacheDbRate());
        int count = 0;
        do {
            if (hangUpSmsConfig.isSendRecordCacheDbInterrupt()) {
                break;
            }
            if (rateLimiter.tryAcquire(1, 100, TimeUnit.MICROSECONDS)) {
                List<UserSendMsgInfo> userSendMsgInfos = hangupMsgSendUserMapper.pageQuery(lid, pageSize, pageIndex);
                returnSize = userSendMsgInfos.size();
                pageIndex += pageSize;
                loadPageData(userSendMsgInfos);
                count += returnSize;
            }

        } while (returnSize >= pageSize);
        LogUtils.info(log, "[LoadCacheFromDb], lid:" + lid + ", total data size:" + count);
    }

    private void loadPageData(List<UserSendMsgInfo> userSendMsgInfos) {
        Map<String, List<LocalDateTime>> phoneTimeMap = new HashMap<>();
        for (UserSendMsgInfo userSendMsgInfo : userSendMsgInfos) {
            List<LocalDateTime> timeList = phoneTimeMap.computeIfAbsent(userSendMsgInfo.getPhonenumber(), k -> new ArrayList<>());
            timeList.add(userSendMsgInfo.getSendmsgtime());
        }
        for (Map.Entry<String, List<LocalDateTime>> entry : phoneTimeMap.entrySet()) {
            saveSendRecordList(entry.getKey(), entry.getValue());
        }
    }

}
